package com.bigdata;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class FlumeInterceptor implements Interceptor {

    Map<String, String> headerMap = new HashMap<>();
    List<Event> eventList = new ArrayList<>();

    @Override
    public void initialize() {

    }

    /**
     * 实现用户的 event 和 商品的 event分别分发到不同的channel
     * user，goods 是从body取值进行判断
     * event header（键值对）中有一个type的key，
     * 对应的value是 first，说明是用户相关的数据，分发到 first channel
     * 对应的value是 second，说明是商品相关的数据，分发到 second channel
     *
     * @param event
     * @return
     */
    @Override
    public Event intercept(Event event) {
        String body = new String(event.getBody());

        if (body.contains("user")) {
            headerMap.put("type", "first");
            event.setHeaders(headerMap);
        } else if (body.contains("goods")) {
            headerMap.put("type", "second");
            event.setHeaders(headerMap);
        } else {
            // 非用户 或者 商品的 日志数据
            // 过滤掉，不往channel传递， return null;
            return null;
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        // list 中的event 清空
        eventList.clear();

        for (Event event : list) {
            Event result = intercept(event);
            if (result != null) {
                eventList.add(result);
            }
        }
        return eventList;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new FlumeInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

}
